package com.fare.taluo.db.hbase;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.stereotype.Component;

import com.dayatang.domain.InstanceFactory;
import com.fare.taluo.vo.core.HBaseFilterVO;

@Component
public class HbaseManager {
	
	protected Log logger= LogFactory.getLog(getClass());
	
	private HbaseDriverManager driverManager;
	
	public HbaseManager(){
		driverManager = HbaseDriverManager.getInstance();
	}
	
	 /**
     * 插入一行记录
     */
    public void addRecord(String tableName, HBaseRowModel model)
            throws Exception{
    	HTableInterface table = null;
    	Put put = null;
        try {
        	if(StringUtils.isNotBlank(tableName) &&
        			model != null){
        	 	put = new Put(Bytes.toBytes(model.getRowkey()));
        	 	//循环将所有值都放入到put中
        		for(HBaseCellModel cell : model.getCells()){
        			if(StringUtils.isNotBlank(cell.getValue())){//写还是不写
        				put.add(Bytes.toBytes(cell.getFamily()),Bytes.toBytes(cell.getQualifier()), Bytes.toBytes(cell.getValue()));
        			}
        		}
	            
	            table = driverManager.getHtable(tableName);
	            table.put(put);
        	}else{
        		if(logger.isDebugEnabled()){
        			logger.debug("传入参数tableName、model为空，没有插入数据!：");
            	}
        	}
        } finally{
        	driverManager.releaseHTable(tableName);
        	table = null;
        	put = null;
        }
    }
    /**
     * 插入单个列记录
     */
    public void addRecord(String tableName,HBaseCellModel cell)
            throws Exception{
    	
    	if(StringUtils.isBlank(cell.getValue())) return;
    	 
    	HTableInterface table = null;
    	Put put = null;
        try {
        	if(StringUtils.isNotBlank(tableName) &&
        			cell != null && cell.getRowModel() != null){
	            put = new Put(Bytes.toBytes(cell.getRowModel().getRowkey()));
	           
	            	put.add(Bytes.toBytes(cell.getFamily()),Bytes.toBytes(cell.getQualifier()), Bytes.toBytes(cell.getValue()));
	            
	            table = driverManager.getHtable(tableName);
	            table.put(put);
        	}else{
        		if(logger.isDebugEnabled()){
        			logger.debug("传入参数tableName、cell为空，没有插入数据!：");
            	}
        	}
        } finally{
        	driverManager.releaseHTable(tableName);
        	table = null;
        	put = null;
        }
    }
 
    /**
     * 删除一行记录
     */
    public void delOneRecord (String tableName, String rowKey) throws Exception{
    	HTableInterface table = null;
    	try{
    		if(StringUtils.isNotBlank(tableName) && StringUtils.isNotBlank(rowKey)){
    			table = driverManager.getHtable(tableName);
    			table.delete(new Delete(rowKey.getBytes()));
    		}else{
        		if(logger.isDebugEnabled()){
        			logger.debug("传入参数tableName、rowkey为空，没有删除数据!：");
            	}
        	}
    	} finally{
    		driverManager.releaseHTable(tableName);
        	table = null;
        }
    }
    
    /**
     * 删除多行记录
     */
    public void delMultRecord (String tableName, String[] rowKeys) throws Exception{
    	HTableInterface table = null;
    	List<Delete> deletes = null;
    	try{
    		if(StringUtils.isNotBlank(tableName) && rowKeys != null && rowKeys.length > 0){
    			table = driverManager.getHtable(tableName);
    			deletes = new ArrayList<Delete>();
    			for(String k : rowKeys){
    				deletes.add(new Delete(k.getBytes()));
    			}
    			
    			table.delete(deletes);
    		}else{
        		if(logger.isDebugEnabled()){
        			logger.debug("传入参数tableName、rowKeys为空，没有删除数据!：");
            	}
        	}
    	} finally{
    		driverManager.releaseHTable(tableName);
        	table = null;
        }
    }
     
    /**
     * 查找一行记录
     * @throws Exception 
     */
    public HBaseRowModel getOneRecord(String tableName, String rowKey) throws Exception{
    	HTableInterface table = null;
    	Get get = null;
    	Result rs = null;
    	HBaseCellModel cell = null;
    	HBaseRowModel model = null;
    	try{
    		model = new HBaseRowModel();
    		if(StringUtils.isNotBlank(tableName) && StringUtils.isNotBlank(rowKey)){
    			table = driverManager.getHtable(tableName);
    			get = new Get(rowKey.getBytes());
    			rs = table.get(get);
    			
    			model.setRowkey(rowKey);
    			
    			for(KeyValue kv : rs.raw()){
    				cell = new HBaseCellModel();
    				cell.setFamily(new String(kv.getFamily()));
    				cell.setQualifier(new String(kv.getQualifier()));
    				cell.setTimestamp(kv.getTimestamp());
    				cell.setValue(new String(kv.getValue()));
    				
    				model.addCell(cell);
    		   }
    		}else{
        		if(logger.isDebugEnabled()){
        			logger.debug("传入参数tableName、rowkey为空，没有删除数据!：");
            	}
        	}
    	} finally{
    		rs = null;
    		get = null;
    		driverManager.releaseHTable(tableName);
        	table = null;
        }
    	return model;
    }
    public Result getResultRecord(String tableName, String rowKey){
    	Result rs = null;
	        try {
	        	Get get = new Get(rowKey.getBytes("utf-8"));
		        HTableInterface table=driverManager.getHtable(tableName);
				rs =table.get(get);
	        table.close();
	        } catch (IOException e) {
				e.printStackTrace();
			}
	        return rs;
    }
    /**
     * 查询多行记录
     * @throws Exception 
     */
    public List<HBaseRowModel> getMultRecord(String tableName, Filter filter) throws Exception {
    	List<HBaseRowModel> results = null;
    	HTableInterface table = null;
    	Scan s = null;
    	ResultScanner ss = null;
    	HBaseCellModel cell = null;
    	HBaseRowModel model = null;
        try{
        	results = new ArrayList<HBaseRowModel>();
              s = new Scan();
              
             if(filter != null) s.setFilter(filter);
             
             table = driverManager.getHtable(tableName);
             ss = table.getScanner(s);
             for(Result r:ss){
            	 //行
            	 model = new HBaseRowModel(new String(r.getRow()));
                 for(KeyValue kv : r.raw()){
                	 //单元格
                	cell = new HBaseCellModel();
                	cell.setFamily(new String(kv.getFamily()));
     				cell.setQualifier(new String(kv.getQualifier()));
     				cell.setTimestamp(kv.getTimestamp());
     				cell.setValue(new String(kv.getValue()));
     				
     				model.addCell(cell);
                 }
                 
                 results.add(model);
             }
            
        } finally{
    		s = null;
    		ss = null;
    		driverManager.releaseHTable(tableName);
        	table = null;
        }
    	return results;
    }
    
    /**
     * 查询多行记录
     * @throws Exception 
     */
    public List<HBaseRowModel> getMultRecord(String tableName, String startRowKey, String endRowKey) throws Exception {
    	List<HBaseRowModel> results = null;
    	HTableInterface table = null;
    	Scan s = null;
    	ResultScanner ss = null;
    	HBaseCellModel cell = null;
    	HBaseRowModel model = null;
        try{
        	results = new ArrayList<HBaseRowModel>();
              s = new Scan();
              s.setStartRow(startRowKey.getBytes());
              s.setStopRow(endRowKey.getBytes());
             table = driverManager.getHtable(tableName);
             ss = table.getScanner(s);
             for(Result r:ss){
            	 //行
            	 model = new HBaseRowModel(new String(r.getRow()));
                 for(KeyValue kv : r.raw()){
                	 //单元格
                	cell = new HBaseCellModel();
                	cell.setFamily(new String(kv.getFamily()));
     				cell.setQualifier(new String(kv.getQualifier()));
     				cell.setTimestamp(kv.getTimestamp());
     				cell.setValue(new String(kv.getValue()));
     				
     				model.addCell(cell);
                 }
                 
                 results.add(model);
             }
            
        } finally{
    		s = null;
    		ss = null;
    		driverManager.releaseHTable(tableName);
        	table = null;
        }
    	return results;
    }
    
    
    /**
     * 查询多行记录，带过滤器
     * @throws Exception 
     */
    public List<HBaseRowModel> getMultRecord(String tableName, String startRowKey, String endRowKey,List<Filter> filterList) throws Exception {
    	List<HBaseRowModel> results = null;
    	HTableInterface table = null;
    	Scan s = null;
    	ResultScanner ss = null;
    	HBaseCellModel cell = null;
    	HBaseRowModel model = null;
        try{
        	results = new ArrayList<HBaseRowModel>();
              s = new Scan();
              s.setStartRow(startRowKey.getBytes());
              s.setStopRow(endRowKey.getBytes());
              //加上过滤器
              if(filterList != null){
            	  for (Filter filter : filterList) {
					s.setFilter(filter);
				}
              }
              // 
            /*  FilterList list = new FilterList(filterList);
  				s.setFilter(list);*/
             table = driverManager.getHtable(tableName);
             ss = table.getScanner(s);
             for(Result r:ss){
            	 //行
            	 model = new HBaseRowModel(new String(r.getRow()));
                 for(KeyValue kv : r.raw()){
                	 //单元格
                	cell = new HBaseCellModel();
                	cell.setFamily(new String(kv.getFamily()));
     				cell.setQualifier(new String(kv.getQualifier()));
     				cell.setTimestamp(kv.getTimestamp());
     				cell.setValue(new String(kv.getValue()));
     				
     				model.addCell(cell);
                 }
                 
                 results.add(model);
             }
            
        } finally{
    		s = null;
    		ss = null;
    		driverManager.releaseHTable(tableName);
        	table = null;
        }
    	return results;
    }
    
    /**
     * 查询多行记录，及返回指定列簇、列、值数据
     * HBaseFilterVO里面包含family,column,value分别对应列簇，列，值,
     * 若一个HBaseFilterVO对象只有family则默认只查询该family,
     * 若带有family和column,则取改family下得column的值
     * 对于List里的多个HBaseFilterVO，关系为逻辑或
     * 			
     * @throws Exception 
     */
    public List<HBaseRowModel> getMultRecordByConditions(String tableName, String startRowKey, 
    		String endRowKey, List<HBaseFilterVO> voList,List<Filter> filters) throws Exception {
    	List<HBaseRowModel> results = null;
    	HTableInterface table = null;
    	Scan s = null;
    	ResultScanner ss = null;
    	HBaseCellModel cell = null;
    	HBaseRowModel model = null;
        try{
        	  results = new ArrayList<HBaseRowModel>();
              s = new Scan();
              s.setStartRow(startRowKey.getBytes());
              s.setStopRow(endRowKey.getBytes());
              if(com.fare.taluo.util.StringUtils.checkList(voList)){
            	  for (HBaseFilterVO vo : voList) {
                	  //没有指定列，则默认指定列簇
                	  if(null == vo.getQualifier() && null != vo.getFamily()){
                		  s.addFamily(vo.getFamily().getBytes());
                	  }else if(null != vo.getQualifier() && null != vo.getFamily()){
                		  s.addColumn(vo.getFamily().getBytes(), vo.getQualifier().getBytes());
                	  }	  
                  }
              }
              
               
            if (com.fare.taluo.util.StringUtils.checkList(filters)) {
      			FilterList list = new FilterList(filters);
      			s.setFilter(list);
      		}
             table = driverManager.getHtable(tableName);
             ss = table.getScanner(s);
             for(Result r:ss){
            	 //行
            	 model = new HBaseRowModel(new String(r.getRow()));
                 for(KeyValue kv : r.raw()){
                	 //单元格
                	cell = new HBaseCellModel();
                	cell.setFamily(new String(kv.getFamily()));
     				cell.setQualifier(new String(kv.getQualifier()));
     				cell.setTimestamp(kv.getTimestamp());
     				cell.setValue(new String(kv.getValue()));
     				
     				model.addCell(cell);
                 }
                 
                 results.add(model);
             }
            
        } finally{
    		s = null;
    		ss = null;
    		driverManager.releaseHTable(tableName);
        	table = null;
        }
    	return results;
    } 
    
    
    public void say(){
    	System.out.println("haha");
    }
    
    
    public static void main(String[] arg) throws Exception{
    	HbaseManager  manage = InstanceFactory.getInstance(HbaseManager.class);
    	manage.say();
    }
}
